package com.ly.kafka.utils;


import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.annotation.PartitionOffset;
import org.springframework.kafka.annotation.TopicPartition;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;
//kafka消费者监听
@Component
public class ConsumerListener {


    /**
     * @Title 指定topic、partition、offset消费
     * @Description 同时监听topic1和topic2，监听topic1的0号分区、topic2的 "0号和1号" 分区，指向1号分区的offset初始值为8
     * @Param [record]
     * @return void
     **/
//    @KafkaListener(id = "consumer1",groupId = "testGroup",topicPartitions = {
//            @TopicPartition(topic = "hello-kraft1", partitions = {"0"}),
//            @TopicPartition(topic = "hello-kraft", partitions = "0",
//                    partitionOffsets = @PartitionOffset(partition = "1", initialOffset = "8"))
//    })
    public void onMessage2(ConsumerRecord<?, ?> record,Acknowledgment ack) {
        System.out.println("topic:"+record.topic()+"|partition:"+record.partition()+"|offset:"+record.offset()+"|value:"+record.value());
        //手动提交offset
        ack.acknowledge();
    }
}
